package io.github.wanggit.antrpc.client.config;

import io.github.wanggit.antrpc.client.zk.IZkClient;
import io.github.wanggit.antrpc.commons.bean.ConfigItem;
import io.github.wanggit.antrpc.commons.bean.ConfigItemData;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.zookeeper.ZkNodeType;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;

import java.util.List;
import java.util.Objects;

@Slf4j
public class DefaultConfigListener implements IConfigListener {

    @Override
    public void doFirstFetch(IConfigContainer configContainer, IZkClient zkClient, String[] paths) {
        if (log.isInfoEnabled()) {
            log.info(
                    "do first fetch configs at application startup. paths="
                            + JsonUtils.toJsonString(paths));
        }
        for (String path : paths) {
            List<String> children = null;
            try {
                children = zkClient.getCurator().getChildren().forPath(path);
            } catch (Exception e) {
                if (log.isWarnEnabled()) {
                    log.warn(
                            "Not Found "
                                    + path
                                    + ", Check it was allowed delayed fetch."
                                    + e.getMessage());
                }
            }
            if (null != children) {
                for (String child : children) {
                    String childFullPath = path + "/" + child;
                    try {
                        byte[] bytes = zkClient.getCurator().getData().forPath(childFullPath);
                        ConfigItem configItem = createConfigItem(childFullPath, bytes);
                        configContainer.saveOrUpdate(configItem);
                    } catch (Exception e) {
                        if (log.isWarnEnabled()) {
                            log.warn(
                                    "Not Found "
                                            + childFullPath
                                            + " data. Check it was allowed delayed fetch."
                                            + e.getMessage());
                        }
                    }
                }
            }
        }
        if (log.isInfoEnabled()) {
            log.info("fetch configs completed at application startup ...");
        }
    }

    @Override
    public void listen(IConfigContainer configContainer, IZkClient zkClient, String[] paths)
            throws ConfigListenException {
        TreeCache cnfTreeCache =
                new TreeCache(zkClient.getCurator(), "/" + ConstantValues.ZK_ROOT_CONFIG_NAME);
        cnfTreeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (Objects.equals(event.getType(), TreeCacheEvent.Type.NODE_ADDED)
                                    || Objects.equals(
                                            event.getType(), TreeCacheEvent.Type.NODE_UPDATED)) {
                                doInternalListen(event, configContainer, paths, false);
                            } else if (Objects.equals(
                                    event.getType(), TreeCacheEvent.Type.NODE_REMOVED)) {
                                doInternalListen(event, configContainer, paths, true);
                            }
                        });
        try {
            cnfTreeCache.start();
        } catch (Exception e) {
            throw new ConfigListenException(e);
        }
    }

    private void doInternalListen(
            TreeCacheEvent event,
            IConfigContainer configContainer,
            String[] paths,
            boolean isDelete) {
        String fullPath = event.getData().getPath();
        boolean needListen = false;
        for (String path : paths) {
            if (fullPath.startsWith(path)) {
                needListen = true;
                break;
            }
        }
        if (needListen) {
            ZkNodeType.ConfigNodeType configNodeType = ZkNodeType.getConfigNodeType(fullPath);
            if (Objects.equals(configNodeType, ZkNodeType.ConfigNodeType.KEY)) {
                ChildData childData = event.getData();
                ConfigItem configItem = createConfigItem(childData.getPath(), childData.getData());
                if (isDelete) {
                    configContainer.delete(configItem);
                } else {
                    configContainer.saveOrUpdate(configItem);
                }
            }
        }
    }

    private ConfigItem createConfigItem(String path, byte[] data) {
        ConfigItem configItem = new ConfigItem();
        configItem.setFullPath(path);
        ConfigItemData configItemData = ConfigItemData.from(data);
        configItem.setKey(configItemData.getKey());
        configItem.setValue(configItemData.getValue());
        configItem.setTs(configItemData.getTs());
        configItem.setGrayReleases(configItemData.getGrayReleases());
        return configItem;
    }
}
